package com.partsoft.dits.mq;

import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

import org.apache.commons.pool.KeyedObjectPool;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import com.ibm.mq.MQC;
import com.ibm.mq.MQException;
import com.ibm.mq.MQQueue;
import com.ibm.mq.MQQueueManager;
import com.partsoft.dits.AbstractConfigurableDescriptorManager;
import com.partsoft.dits.AbstractMessageByteBatchProcessor;
import com.partsoft.dits.DataEdgeDescriptor;
import com.partsoft.dits.DataIntegrateBatch;
import com.partsoft.dits.DataMessageReceiver;
import com.partsoft.dits.DataReceiver;
import com.partsoft.dits.DescriptorAccepter;
import com.partsoft.dits.DescriptorDepends;
import com.partsoft.dits.DescriptorManager;
import com.partsoft.dits.DescriptorRegistry;
import com.partsoft.dits.EdgeDescriptor;
import com.partsoft.dits.MonitorStartMode;
import com.partsoft.dits.log.Log;
import com.partsoft.dits.message.MessageEntity;
import com.partsoft.dits.message.MessageException;
import com.partsoft.dits.message.MessageProcessor;

/**
 * 上下文可配置的MQ队列监听引擎
 */
@SuppressWarnings("unchecked")
public class ContextConfigurableMQQueueMonitorReactor extends
		AbstractConfigurableDescriptorManager<MQQueueMonitorDescriptor> implements
		MQQueueMonitorReactor, DescriptorManager<MQQueueMonitorDescriptor>, 
		DescriptorRegistry<MQQueueMonitorDescriptor>,
		DataMessageReceiver,
		ApplicationListener<ApplicationEvent>, 
		ApplicationContextAware, 
		DescriptorDepends<MQQueueMonitorDescriptor> {

	private Set<Class<? extends EdgeDescriptor>> dependsClasses;
	{
		dependsClasses = new HashSet<Class<? extends EdgeDescriptor>>(1);
		dependsClasses.add(MQPoolDescriptor.class);
	}
	
	
	private MQPoolProvider mqPoolProvider;
	
	private DataReceiver messageEntityReceiver;
	
	private Map<String, MQQueueListenerWorkerRef> workers = Collections.synchronizedMap(new HashMap<String, MQQueueListenerWorkerRef>(5));
	
	private MessageProcessor messageProcessor = MessageProcessor.INSTANCE;
	
	private AbstractMessageByteBatchProcessor msgProcessor = new AbstractMessageByteBatchProcessor() {
		
		@Override
		protected Object doProcess(DataEdgeDescriptor dataDescriptor, byte[] bytes, DescriptorAccepter<EdgeDescriptor> accepter) {
			try {
				ContextConfigurableMQQueueMonitorReactor.this.doProcess((MQQueueMonitorDescriptor) dataDescriptor, bytes, accepter);
			} catch (MessageException e) {
				throw new IllegalArgumentException(e.getMessage(), e);
			}
			return null;
		}
		
		protected void validateBatch(DataIntegrateBatch batch) {
			super.validateBatch(batch);
			validateSupportDataDescriptor(batch.getDataDescriptor());
		};
	};
	
	public DataReceiver getMessageEntityReceiver() {
		return messageEntityReceiver;
	}

	public void setMessageEntityReceiver(DataReceiver messageEntityReceiver) {
		this.messageEntityReceiver = messageEntityReceiver;
	}

	public void setMessageProcessor(MessageProcessor messageProcessor) {
		this.messageProcessor = messageProcessor;
	}
	
	public MessageProcessor getMessageProcessor() {
		return messageProcessor;
	}
	
	public void setMQPoolProvider(MQPoolProvider mqPoolProvider) {
		assertNoInit();
		this.mqPoolProvider = mqPoolProvider;
	}
	
	public MQPoolProvider getMQPoolProvider() {
		return mqPoolProvider;
	}
	
	public Class<MQQueueMonitorDescriptor> getSupportClass() {
		return MQQueueMonitorDescriptor.class;
	}
	
	private void createAndStartMQQueueMonitor(MQQueueMonitorDescriptor descriptor) {
		synchronized (this) {
			if (descriptor.getStartMode() == MonitorStartMode.DISABLED) return;
			String monitorWorkerID = resolveMonitorWorkerID(descriptor);
			MQQueueListenerWorkerRef workerRef = null;
			if (workers.containsKey(monitorWorkerID)) {
				workerRef = workers.get(monitorWorkerID);
			} else {
				KeyedObjectPool qmPool = getMQPoolProvider().getMQPool(descriptor.getConnectPool());
				MQQueueListenerContextImpl ctx = new MQQueueListenerContextImpl(this, descriptor.getConnectPool(), qmPool, descriptor.getQueueManager(), descriptor.getQueue(), descriptor.getMonitorInterval());
				MQQueueListenerWorker worker = new MQQueueListenerWorker(ctx);
				workerRef = new MQQueueListenerWorkerRef(worker);
				workers.put(monitorWorkerID, workerRef);
				Thread thread = new Thread(workerRef, String.format("MQMonitor[%s-%s-%s]", descriptor.getConnectPool(), descriptor.getQueueManager(), descriptor.getQueue()));
				thread.start();
			}
			workerRef.addMonitorDescriptor(descriptor);
			workerRef.addRef();
			
		}
	}
	
	private void stopAndRemoveMQQueueMonitor(MQQueueMonitorDescriptor descriptor) {
		synchronized (this) {
			String monitorWorkerID = resolveMonitorWorkerID(descriptor);
			if (!workers.containsKey(monitorWorkerID)) return;
			MQQueueListenerWorkerRef workerRef = workers.get(monitorWorkerID);
			workerRef.removeMonitorDescriptor(descriptor);
			if (workerRef.removeRef() == 0) {
				workerRef.stop();
				workers.remove(monitorWorkerID);
			}
		}
	}

	@Override
	protected void beforeRemoveDescriptor(MQQueueMonitorDescriptor descriptor) {
		stopAndRemoveMQQueueMonitor(descriptor);
	}

	@Override
	protected void postRemoveDescriptor(MQQueueMonitorDescriptor descriptor) {}

	@Override
	protected void beforeLoadConfiguredDescriptor(MQQueueMonitorDescriptor descriptor) {}
	
	@Override
	protected void postLoadConfiguredDescriptor(MQQueueMonitorDescriptor descriptor) {
		createAndStartMQQueueMonitor(descriptor);
	}
	
	@Override
	protected void setupDescriptorIdentifier(MQQueueMonitorDescriptor descriptor, String id) {
		descriptor.setId(id);
	}

	@Override
	protected Properties beforeAddDescriptor(MQQueueMonitorDescriptor descriptor) {
		Properties props = new Properties();
		props.setProperty("connectPool", descriptor.getConnectPool());
		props.setProperty("queueManager", descriptor.getQueueManager());
		props.setProperty("queue", descriptor.getQueue());
		props.setProperty("monitorInterval", Long.toString(descriptor.getMonitorInterval()));
		props.setProperty("description", descriptor.getDescription());
		if (descriptor.getDataClass() != null)  props.setProperty("dataClass", descriptor.getDataClass().getName());
		props.setProperty("title", descriptor.getTitle());
		if (descriptor.getDataDescriptorId() != null) props.setProperty("dataDescriptorId", descriptor.getDataDescriptorId());
		if (StringUtils.hasText(descriptor.getMessageProcessorClazz())) {
			props.setProperty("messageProcessorClazz", descriptor.getMessageProcessorClazz());
		}
		props.setProperty("ordinal", Integer.valueOf(descriptor.getOrdinal()).toString());

		return props;
	}
	
	@Override
	protected void postAddDescriptor(MQQueueMonitorDescriptor descriptor) {
		createAndStartMQQueueMonitor(descriptor);
	}

	@Override
	protected void validateDescriptor(MQQueueMonitorDescriptor descriptor) {
		Assert.isTrue(descriptor.getMonitorInterval() > 0, "监听间隔时间必须大于0秒。");
		Assert.hasText(descriptor.getConnectPool(), "MQ池名称属性必须设置。");
		Assert.hasText(descriptor.getQueueManager(), "监听队列管理器名称必须设置。");
		Assert.hasText(descriptor.getQueue(), "监听队列名称必须设置");
		
		//校验MQ池
		Assert.isTrue(getMQPoolProvider().hasMQPool(descriptor.getConnectPool()), String.format("找不到指定的MQ池(%s)。", descriptor.getConnectPool()));
		KeyedObjectPool qmPool = getMQPoolProvider().getMQPool(descriptor.getConnectPool());
		Object qmObject = null;
		try {
			qmObject = qmPool.borrowObject(descriptor.getQueueManager());
		} catch (Throwable e) {
			throw new IllegalArgumentException(String.format("MQ池(%s)中找不到队列管理器(%s)", descriptor.getConnectPool(), descriptor.getQueueManager()), e);
		}
		MQQueueManager qmgr = (MQQueueManager) qmObject;
		MQQueue queue = null;
		try {
			queue = qmgr.accessQueue(descriptor.getQueue(), MQC.MQOO_INPUT_SHARED | MQC.MQOO_OUTPUT | MQC.MQOO_FAIL_IF_QUIESCING);
		} catch (MQException e) {
			throw new IllegalArgumentException(String.format("打开MQ池(%s)队列管理器(%s)中的队列(%s)出错。", descriptor.getConnectPool(), descriptor.getQueueManager(), descriptor.getQueue()), e);
		} finally {
			if (queue != null) {
				try {
					queue.close();
				} catch (MQException e) {}
			}
			try {
				qmPool.returnObject(descriptor.getQueueManager(), qmgr);
			} catch (Exception e) {
				throw new IllegalArgumentException(e.getMessage(), e);
			}
		}
	}
	
	protected String resolveMonitorWorkerID(MQQueueMonitorDescriptor descriptor) {
		StringBuffer buffer = new StringBuffer();
		buffer.append(descriptor.getConnectPool()).append(".").
			append(descriptor.getQueueManager()).append(".").
			append(descriptor.getQueue()).append(".");
		return buffer.toString();
	}

	@Override
	protected String resolveDescriptorIdentifier(MQQueueMonitorDescriptor descriptor) {
		if (StringUtils.hasText(descriptor.getId())) return descriptor.getId();
		StringBuffer buffer = new StringBuffer();
		buffer.append(descriptor.getConnectPool()).append(".").
			append(descriptor.getQueueManager()).append(".").
			append(descriptor.getQueue()).append(".").
			append(descriptor.getDataClass() != null ? descriptor.getDataClass().getName() : "nodata").append(".").
			append(descriptor.getDataDescriptorId());
		return buffer.toString();
	}
	
	@Override
	public void initContext() {
		Assert.notNull(getMQPoolProvider(), "必须设置MQPoolProvider。");
		Assert.notNull(getMessageProcessor(), "必须设置消息处理器。");
		Assert.notNull(getMessageEntityReceiver(), "必须设置数据处理器。");
		super.initContext();
	}
	
	@Override
	protected void beforeUnLoadConfiguredDescriptor(MQQueueMonitorDescriptor descriptor) {
		stopAndRemoveMQQueueMonitor(descriptor);
	}
	
	public boolean supportData(Class<? extends DataEdgeDescriptor> descriptorClazz) {
		return ClassUtils.isAssignable(getSupportClass(), descriptorClazz);
	}

	public boolean supportData(DataEdgeDescriptor dataDescriptor) {
		return supportData((Class<? extends DataEdgeDescriptor>) ClassUtils.getUserClass(dataDescriptor)) ?
				hasDescriptor(dataDescriptor.getId()) : false;
	}
	
	protected Class<? extends DataEdgeDescriptor> validateSupportDataDescriptor(DataEdgeDescriptor descriptor) {
		Class<? extends DataEdgeDescriptor> descriptorClazz = (Class<? extends DataEdgeDescriptor>) ClassUtils.getUserClass(descriptor);
		Assert.isTrue(supportData(descriptor), String.format("不支持此数据(%s)，描述符(%s)。", descriptorClazz.getName(), descriptor.getId()));
		return descriptorClazz;
	}

	public void process(DataIntegrateBatch batch) {
		msgProcessor.process(batch);
	}
	
	protected void doProcess(MQQueueMonitorDescriptor dataDescriptor, byte[] databytes, DescriptorAccepter<EdgeDescriptor> accepter) throws MessageException {
		MessageProcessor mpr = getMessageProcessor();
		if (StringUtils.hasText(dataDescriptor.getMessageProcessorClazz())) {
			//TODO 实现动态加载消息解码器内容。
			//TODO 需要实现过滤
			throw new IllegalStateException("尚未实现。");
		}
		MessageEntity msgEntity = null;
		try {
			msgEntity = mpr.fromMessageBytes(databytes);
		} catch (MessageException e) {
			throw e;
		}
		
		if (Log.isDebugEnabled()) Log.debug(String.format("解码消息(%s)成功，数据编号(%s)，准备处理...", msgEntity.getHeader().getMessageCode(), msgEntity.getHeader().getMessageId()));
		try {
			getMessageEntityReceiver().receiveData(msgEntity);
		} catch (Throwable e) {
			throw new MessageException(e.getMessage(), databytes, e);
		}
	}
	
	@Override
	public Set<Class<? extends EdgeDescriptor>> getDependClasses() {
		Set<Class<? extends EdgeDescriptor>> clazzes = new LinkedHashSet<Class<? extends EdgeDescriptor>>(dependsClasses);
		for (MQQueueMonitorDescriptor desc : getDescriptorMap().values()) {
			if (!clazzes.contains(desc.getDataClass()))
				clazzes.add(desc.getDataClass());
		}
		return Collections.unmodifiableSet(new HashSet<Class<? extends EdgeDescriptor>>(clazzes));
	}

	@Override
	public boolean dependDescriptor(EdgeDescriptor descriptor) {
		boolean result = false;
		for (MQQueueMonitorDescriptor desc : getDescriptorMap().values()) {
			if (ClassUtils.isAssignableValue(MQPoolDescriptor.class, descriptor)) {
				result = ObjectUtils.nullSafeEquals(((MQPoolDescriptor)descriptor).getId(), desc.getConnectPool());
			} else if (ClassUtils.isAssignableValue(DataEdgeDescriptor.class, descriptor) && 
					ClassUtils.isAssignableValue(desc.getDataClass(), descriptor)) {
				result = ObjectUtils.nullSafeEquals(((DataEdgeDescriptor)descriptor).getId(), desc.getDataDescriptorId());
			}
			if (result) break;
		}
		return result;
	}
	
	@Override
	public MQQueueMonitorDescriptor[] getDescriptorByDepend(EdgeDescriptor descriptor) {
		MQQueueMonitorDescriptor []returnArray = new MQQueueMonitorDescriptor[0];
		if (dependDescriptor(descriptor)) {
			List<MQQueueMonitorDescriptor> returnList = new LinkedList<MQQueueMonitorDescriptor>();
			for (MQQueueMonitorDescriptor desc : getDescriptorMap().values()) {
				if (ClassUtils.isAssignableValue(MQPoolDescriptor.class, descriptor) &&
						ObjectUtils.nullSafeEquals(((MQPoolDescriptor)descriptor).getId(), desc.getConnectPool())) {
					returnList.add(desc);
				} else if (ClassUtils.isAssignableValue(DataEdgeDescriptor.class, descriptor) && 
						ClassUtils.isAssignableValue(desc.getDataClass(), descriptor) && 
						ObjectUtils.nullSafeEquals(((DataEdgeDescriptor)descriptor).getId(), desc.getDataDescriptorId())) {
					returnList.add(desc);
				}
			}
			returnArray = returnList.toArray(returnArray);
		}
		return returnArray;
	}

	public void receiveData(DataEdgeDescriptor dataDescriptor, byte[] databytes) {
		validateSupportDataDescriptor(dataDescriptor);
		try {
			doProcess((MQQueueMonitorDescriptor) dataDescriptor, databytes, null);
		} catch (MessageException e) {
			throw new IllegalArgumentException(e.getMessage(), e);
		}
	}

	static class MQQueueListenerWorkerRef implements Runnable {
		
		private final MQQueueListenerWorker worker;
		
		private int ref = 0;
		
		MQQueueListenerWorkerRef(MQQueueListenerWorker worker) {
			this.worker = worker;
		}
		
		public void addMonitorDescriptor(MQQueueMonitorDescriptor desc) {
			getListenerContext().addMonitorDescriptor(desc);
		}
		
		public void removeMonitorDescriptor(DataEdgeDescriptor desc) {
			getListenerContext().removeMonitorDescriptor(desc);
		}
		
		private MQQueueListenerContextImpl getListenerContext() {
			return worker.getListenerContext();
		}
		
		public void run() {
			worker.run();
		}
		
		public void stop() {
			worker.stop();
		}
		
		public int addRef() {
			return ++ref;
		}
		
		public int removeRef() {
			return --ref;
		}
	}
	
	static class MQQueueListenerContextImpl implements MQQueueListenerContext {
		
		private final DataMessageReceiver dataMessageReceiver;
		
		private final long retrieveInterval;
		
		private final String listenerQueueName;
		
		private final String listenerQMName;
		
		private final KeyedObjectPool qmPool;
		
		private final String mqConnectPool;
		
		private Set<MQQueueMonitorDescriptor> dataDescriptors = Collections.synchronizedSet(new HashSet<MQQueueMonitorDescriptor>(2));
		
		MQQueueListenerContextImpl(DataMessageReceiver dataMessageReceiver, String mqConnectPool, KeyedObjectPool qmPool, String listenerQMName, String listenerQueueName, long retrieveInterval) {
			this.mqConnectPool = mqConnectPool;
			this.dataMessageReceiver = dataMessageReceiver;
			this.qmPool = qmPool;
			this.listenerQMName = listenerQMName;
			this.listenerQueueName = listenerQueueName;
			this.retrieveInterval = retrieveInterval;
		}
		
		public EdgeDescriptor getEdgeDescriptor() {
			return new MQPoolDescriptor(mqConnectPool);
		}

		
		public MQQueueMonitorDescriptor[] getDataDescriptors() {
			return dataDescriptors.toArray(new MQQueueMonitorDescriptor[0]);
		}
		
		public void addMonitorDescriptor(MQQueueMonitorDescriptor desc) {
			dataDescriptors.add(desc);
		}
		
		public void removeMonitorDescriptor(DataEdgeDescriptor desc) {
			dataDescriptors.remove(desc);
		}
		
		public void receiveData(DataEdgeDescriptor desc, byte[] sourceXML) {
			Assert.notNull(sourceXML);
			dataMessageReceiver.receiveData(desc, sourceXML);
		}

		public KeyedObjectPool getQueueManagerPool() {
			return qmPool;
		}

		public String getQueueManager() {
			return listenerQMName;
		}

		public String getQueue() {
			return listenerQueueName;
		}

		public long getRetrieveInterval() {
			return retrieveInterval;
		}
	}
}
